package com.qyer.commons.concurrent.assemblingline;

import com.qyer.commons.CheckStatus;
import com.qyer.commons.concurrent.ArrayListBulk;
import com.qyer.commons.concurrent.GenericLinkedBlockingQueue;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Closeable;

/**
 * User: Z J Wu Date: 2016/06/01 Time: 19:27 Package: com.qyer.commons.queueprocess
 */
public abstract class RawDataGenerator<T> implements Closeable, AssembleBulkConveyors<T>,
  AssembleBulkNotifier<T>, CheckStatus {

  private static final Logger LOGGER = LoggerFactory.getLogger(RawDataGenerator.class);

  protected GenericLinkedBlockingQueue<ArrayListBulk<T>> conveyor;
  protected StatisticContext context;
  protected Notifier<ArrayListBulk<T>> notifier;
  protected String name;
  protected AssemblingLineStatus status;
  protected int bulkSize;

  public RawDataGenerator(String name, StatisticContext context) {
    this.name = name;
    this.context = context;
  }

  protected abstract void beginProduce(
    GenericLinkedBlockingQueue<ArrayListBulk<T>> mainQueue) throws Exception;

  public void beginJob() throws Exception {
    LOGGER.info("Raw data generator started.");
    beginProduce(this.conveyor);
    LOGGER.info("All data has been produced. Begin to put poison pill({}).",
                notifier.getNextConveyorPillCount());
    notifier.finishAndNotify();
    LOGGER.info("All raw data conveyor pills have been put.");
  }

  public Notifier<ArrayListBulk<T>> getNotifier() {
    return notifier;
  }

  public String getName() {
    return name;
  }

  public StatisticContext getContext() {
    return context;
  }

  @Override
  public void assembleConveyors(GenericLinkedBlockingQueue<ArrayListBulk<T>>... conveyors) {
    if (ArrayUtils.isNotEmpty(conveyors)) {
      this.conveyor = conveyors[0];
      this.status = check();
    }
  }

  @Override
  public void assembleNotifier(Notifier<ArrayListBulk<T>> notifier) {
    if (notifier != null) {
      this.notifier = notifier;
      this.status = check();
    }
  }

  public void setBulkSize(int bulkSize) {
    this.bulkSize = bulkSize;
  }

  @Override
  public String toString() {
    return new ToStringBuilder(this, ToStringStyle.SHORT_PREFIX_STYLE).append("name", name)
                                                                      .append("conveyor", conveyor)
                                                                      .append("notifier", notifier)
                                                                      .append("status", status)
                                                                      .toString();
  }

  @Override
  public AssemblingLineStatus check() {
    if (StringUtils.isEmpty(name) || conveyor == null || context == null || notifier == null) {
      return AssemblingLineStatus.INITED;
    }
    return AssemblingLineStatus.STANDBY;
  }
}
